package org.example.flinktest.apidemo.sinktest;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.example.flinktest.bean.SensorReading;
import org.example.flinktest.operators.MakeSensorMapFunction;

/**
 * @author shihb
 * @date 2019/12/11 12:09
 */
public class JdbcSinkDemo {

  public static void main(String[] args) throws Exception {
    //环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);

    //source
    DataStream<String> input = env.readTextFile(
        "D:\\SHBData\\IDEAProjects\\fink-parent\\flink-quickstart-java\\src\\main\\resources\\sensor.txt");



    //transform操作
    DataStream<SensorReading> dataStream = input.map(new MakeSensorMapFunction());


    //sink
    dataStream.addSink(new MyJdbcSink());


    env.execute("jdbc sink test");
  }

}
class  MyJdbcSink extends RichSinkFunction<SensorReading>{
  //定义sql的连接，预编译器

  private Connection conn;
  private PreparedStatement insertStmt;
  private PreparedStatement updateStmt;

  @Override
  /**
   *初始化连接和预编译语句
   */
  public void open(Configuration parameters) throws Exception {
    super.open(parameters);
    Class.forName("com.gbase.jdbc.Driver");
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test","user","123456");
    insertStmt = conn.prepareStatement("INSERT INTO temperatures(sensor,temp) VALUES(?,?)");
    updateStmt = conn.prepareStatement("UPDATE temperatures SET temp = ? WHERE sensor = ? ");
  }

  @Override
  /**
   *调用连接，执行sql
   */
  public void invoke(SensorReading value, Context context) throws Exception {
    //执行更新语句
    updateStmt.setDouble(1,value.getTemperature());
    updateStmt.setString(2,value.getId());
    updateStmt.execute();
    //如果key不存在，执行插入语句
    if(updateStmt.getUpdateCount()==0){
      insertStmt.setString(1,value.getId());
      insertStmt.setDouble(2,value.getTemperature());
      insertStmt.execute();
    }

  }

  @Override

  /**
   *关闭是做一些清理工作
   */
  public void close() throws Exception {
    insertStmt.close();
    updateStmt.close();
    conn.close();
  }
}
